package com.me.bigdata

import java.net.URL

import org.apache.spark.{SparkConf, SparkContext}

/**
  * 计算每门学科最受欢迎的老师
  * 日志格式：
  * http://java.xxx.com/zhang
  * http://java.xxx.com/zhang
  * http://java.xxx.com/li
  * http://scala.xxx.com/wang
  * http://scala.xxx.com/wang
  */
object ScalaSubjectTeacherCount {

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local[*]").setAppName("ScalaSubjectTeacherCount")
    val sc = new SparkContext(conf)
    val lines = sc.textFile(args(0), 5)
    val result = lines.map(line => {
      val host = new URL(line).getHost
      val subject = host.substring(0, host.indexOf("."))
      val teacher = line.substring(line.lastIndexOf("/") + 1, line.length)
      (subject, teacher)
    }).map((_, 1)).reduceByKey(_+_).groupBy(_._1._1).mapValues(v => v.toList.sortBy(_._2).reverse.take(2))
    result.saveAsTextFile(args(1))
  }
}
